/*-------------------------------------------------------------------------
 *
 * ginfast.c
 *	  Postgres 倒排索引访问方法的快速插入例程。
 *	  待处理的条目存储在线性页面列表中。
 *	  稍后（通常在 VACUUM 期间），将调用 ginInsertCleanup()
 *	  将待处理的条目转移到常规索引结构中。
 *	  这会带来优势，因为批量插入比逐条插入高效得多。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *			src/backend/access/gin/ginfast.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/gin_private.h"
#include "access/ginxlog.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "catalog/pg_am.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "port/pg_bitutils.h"
#include "postmaster/autovacuum.h"
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "storage/predicate.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/rel.h"

/* GUC 参数 */
int			gin_pending_list_limit = 0;

#define GIN_PAGE_FREESIZE \
	( BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(GinPageOpaqueData)) )

typedef struct KeyArray
{
	Datum	   *keys;			/* 可扩展数组 */
	GinNullCategory *categories;	/* 另一个可扩展数组 */
	int32		nvalues;		/* 当前有效条目的数量 */
	int32		maxvalues;		/* 数组的分配大小 */
} KeyArray;


/*
 * 从给定的元组数组构建一个待处理列表页面，并将其写出。
 *
 * 返回页面上剩余的空闲空间量。
 */
static int32 fc_writeListPage(Relation fc_index, Buffer fc_buffer,
			  IndexTuple *fc_tuples, int32 fc_ntuples, BlockNumber fc_rightlink)
{
	Page		fc_page = BufferGetPage(fc_buffer);
	int32		fc_i,
				fc_freesize,
				fc_size = 0;
	OffsetNumber fc_l,
				fc_off;
	PGAlignedBlock fc_workspace;
	char	   *fc_ptr;

	START_CRIT_SECTION();

	GinInitBuffer(fc_buffer, GIN_LIST);

	fc_off = FirstOffsetNumber;
	fc_ptr = fc_workspace.data;

	for (fc_i = 0; fc_i < fc_ntuples; fc_i++)
	{
		int			fc_this_size = IndexTupleSize(fc_tuples[fc_i]);

		memcpy(fc_ptr, fc_tuples[fc_i], fc_this_size);
		fc_ptr += fc_this_size;
		fc_size += fc_this_size;

		fc_l = PageAddItem(fc_page, (Item) fc_tuples[fc_i], fc_this_size, fc_off, false, false);

		if (fc_l == InvalidOffsetNumber)
			elog(ERROR, "failed to add item to index page in \"%s\"",
				 RelationGetRelationName(fc_index));

		fc_off++;
	}

	Assert(fc_size <= BLCKSZ);		/* 否则我们会溢出工作区 */

	GinPageGetOpaque(fc_page)->rightlink = fc_rightlink;

	/*
	 * 尾页可能只包含完整的行或放置在前面的页面上的行的最终部分
	 * （这里的“行”指的是为一个堆元组生成的所有索引元组）
	 */
	if (fc_rightlink == InvalidBlockNumber)
	{
		GinPageSetFullRow(fc_page);
		GinPageGetOpaque(fc_page)->maxoff = 1;
	}
	else
	{
		GinPageGetOpaque(fc_page)->maxoff = 0;
	}

	MarkBufferDirty(fc_buffer);

	if (RelationNeedsWAL(fc_index))
	{
		ginxlogInsertListPage fc_data;
		XLogRecPtr	fc_recptr;

		fc_data.rightlink = fc_rightlink;
		fc_data.ntuples = fc_ntuples;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_data, sizeof(ginxlogInsertListPage));

		XLogRegisterBuffer(0, fc_buffer, REGBUF_WILL_INIT);
		XLogRegisterBufData(0, fc_workspace.data, fc_size);

		fc_recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_INSERT_LISTPAGE);
		PageSetLSN(fc_page, fc_recptr);
	}

	/* 在释放缓冲区之前获取空闲空间 */
	fc_freesize = PageGetExactFreeSpace(fc_page);

	UnlockReleaseBuffer(fc_buffer);

	END_CRIT_SECTION();

	return fc_freesize;
}

static void fc_makeSublist(Relation fc_index, IndexTuple *fc_tuples, int32 fc_ntuples,
			GinMetaPageData *fc_res)
{
	Buffer		fc_curBuffer = InvalidBuffer;
	Buffer		fc_prevBuffer = InvalidBuffer;
	int			fc_i,
				fc_size = 0,
				fc_tupsize;
	int			fc_startTuple = 0;

	Assert(fc_ntuples > 0);

	/*
	 * 将元组拆分为页面
	 */
	for (fc_i = 0; fc_i < fc_ntuples; fc_i++)
	{
		if (fc_curBuffer == InvalidBuffer)
		{
			fc_curBuffer = GinNewBuffer(fc_index);

			if (fc_prevBuffer != InvalidBuffer)
			{
				fc_res->nPendingPages++;
				fc_writeListPage(fc_index, fc_prevBuffer,
							  fc_tuples + fc_startTuple,
							  fc_i - fc_startTuple,
							  BufferGetBlockNumber(fc_curBuffer));
			}
			else
			{
				fc_res->head = BufferGetBlockNumber(fc_curBuffer);
			}

			fc_prevBuffer = fc_curBuffer;
			fc_startTuple = fc_i;
			fc_size = 0;
		}

		fc_tupsize = MAXALIGN(IndexTupleSize(fc_tuples[fc_i])) + sizeof(ItemIdData);

		if (fc_size + fc_tupsize > GinListPageSize)
		{
			/* 不会适合，强制创建新页面并重新处理 */
			fc_i--;
			fc_curBuffer = InvalidBuffer;
		}
		else
		{
			fc_size += fc_tupsize;
		}
	}

	/*
	 * 写入最后一页
	 */
	fc_res->tail = BufferGetBlockNumber(fc_curBuffer);
	fc_res->tailFreeSize = fc_writeListPage(fc_index, fc_curBuffer,
									  fc_tuples + fc_startTuple,
									  fc_ntuples - fc_startTuple,
									  InvalidBlockNumber);
	fc_res->nPendingPages++;
	/* 那仅仅是一个堆元组 */
	fc_res->nPendingHeapTuples = 1;
}

/*
 * 将 *collector 中包含的索引元组写入索引的待处理列表中。
 *
 * 函数保证所有这些元组将按顺序连续插入，
 * 保持顺序
 */
void ginHeapTupleFastInsert(GinState *fc_ginstate, GinTupleCollector *fc_collector)
{
	Relation	fc_index = fc_ginstate->index;
	Buffer		fc_metabuffer;
	Page		fc_metapage;
	GinMetaPageData *fc_metadata = NULL;
	Buffer		fc_buffer = InvalidBuffer;
	Page		fc_page = NULL;
	ginxlogUpdateMeta fc_data;
	bool		fc_separateList = false;
	bool		fc_needCleanup = false;
	int			fc_cleanupSize;
	bool		fc_needWal;

	if (fc_collector->ntuples == 0)
		return;

	fc_needWal = RelationNeedsWAL(fc_index);

	fc_data.node = fc_index->rd_node;
	fc_data.ntuples = 0;
	fc_data.newRightlink = fc_data.prevTail = InvalidBlockNumber;

	fc_metabuffer = ReadBuffer(fc_index, GIN_METAPAGE_BLKNO);
	fc_metapage = BufferGetPage(fc_metabuffer);

	/*
	 * 对待处理列表的插入在逻辑上可能属于树中的任何地方，
	 * 因此它与所有可串行化扫描发生冲突。所有扫描都在元缓冲区上获取
	 * 谓词锁以表示这一点。因此，我们将在持有页面锁并准备修改页面之前
	 * 检查冲突。
	 */

	if (fc_collector->sumsize + fc_collector->ntuples * sizeof(ItemIdData) > GinListPageSize)
	{
		/*
		 * 总大小大于一页 => 创建子列表
		 */
		fc_separateList = true;
	}
	else
	{
		LockBuffer(fc_metabuffer, GIN_EXCLUSIVE);
		fc_metadata = GinPageGetMeta(fc_metapage);

		if (fc_metadata->head == InvalidBlockNumber ||
			fc_collector->sumsize + fc_collector->ntuples * sizeof(ItemIdData) > fc_metadata->tailFreeSize)
		{
			/*
			 * 待处理列表为空或总大小大于尾页的空闲空间 =>
			 * 创建子列表
			 *
			 * 我们解锁元缓冲区以保持高并发
			 */
			fc_separateList = true;
			LockBuffer(fc_metabuffer, GIN_UNLOCK);
		}
	}

	if (fc_separateList)
	{
		/*
		 * 我们应该单独创建子列表并将其附加到尾部
		 */
		GinMetaPageData fc_sublist;

		memset(&fc_sublist, 0, sizeof(GinMetaPageData));
		fc_makeSublist(fc_index, fc_collector->tuples, fc_collector->ntuples, &fc_sublist);

		/*
		 * 元页面已经解锁，见上文
		 */
		LockBuffer(fc_metabuffer, GIN_EXCLUSIVE);
		fc_metadata = GinPageGetMeta(fc_metapage);

		CheckForSerializableConflictIn(fc_index, NULL, GIN_METAPAGE_BLKNO);

		if (fc_metadata->head == InvalidBlockNumber)
		{
			/*
			 * 主列表为空，因此只需将子列表插入为主列表
			 */
			START_CRIT_SECTION();

			fc_metadata->head = fc_sublist.head;
			fc_metadata->tail = fc_sublist.tail;
			fc_metadata->tailFreeSize = fc_sublist.tailFreeSize;

			fc_metadata->nPendingPages = fc_sublist.nPendingPages;
			fc_metadata->nPendingHeapTuples = fc_sublist.nPendingHeapTuples;

			if (fc_needWal)
				XLogBeginInsert();
		}
		else
		{
			/*
			 * 合并列表
			 */
			fc_data.prevTail = fc_metadata->tail;
			fc_data.newRightlink = fc_sublist.head;

			fc_buffer = ReadBuffer(fc_index, fc_metadata->tail);
			LockBuffer(fc_buffer, GIN_EXCLUSIVE);
			fc_page = BufferGetPage(fc_buffer);

			Assert(GinPageGetOpaque(fc_page)->rightlink == InvalidBlockNumber);

			START_CRIT_SECTION();

			GinPageGetOpaque(fc_page)->rightlink = fc_sublist.head;

			MarkBufferDirty(fc_buffer);

			fc_metadata->tail = fc_sublist.tail;
			fc_metadata->tailFreeSize = fc_sublist.tailFreeSize;

			fc_metadata->nPendingPages += fc_sublist.nPendingPages;
			fc_metadata->nPendingHeapTuples += fc_sublist.nPendingHeapTuples;

			if (fc_needWal)
			{
				XLogBeginInsert();
				XLogRegisterBuffer(1, fc_buffer, REGBUF_STANDARD);
			}
		}
	}
	else
	{
		/*
		 * 插入到尾页。元页面已经锁定
		 */
		OffsetNumber fc_l,
					fc_off;
		int			fc_i,
					fc_tupsize;
		char	   *fc_ptr;
		char	   *fc_collectordata;

		CheckForSerializableConflictIn(fc_index, NULL, GIN_METAPAGE_BLKNO);

		fc_buffer = ReadBuffer(fc_index, fc_metadata->tail);
		LockBuffer(fc_buffer, GIN_EXCLUSIVE);
		fc_page = BufferGetPage(fc_buffer);

		fc_off = (PageIsEmpty(fc_page)) ? FirstOffsetNumber :
			OffsetNumberNext(PageGetMaxOffsetNumber(fc_page));

		fc_collectordata = fc_ptr = (char *) palloc(fc_collector->sumsize);

		fc_data.ntuples = fc_collector->ntuples;

		START_CRIT_SECTION();

		if (fc_needWal)
			XLogBeginInsert();

		/*
		 * 增加堆元组的计数器
		 */
		Assert(GinPageGetOpaque(fc_page)->maxoff <= fc_metadata->nPendingHeapTuples);
		GinPageGetOpaque(fc_page)->maxoff++;
		fc_metadata->nPendingHeapTuples++;

		for (fc_i = 0; fc_i < fc_collector->ntuples; fc_i++)
		{
			fc_tupsize = IndexTupleSize(fc_collector->tuples[fc_i]);
			fc_l = PageAddItem(fc_page, (Item) fc_collector->tuples[fc_i], fc_tupsize, fc_off, false, false);

			if (fc_l == InvalidOffsetNumber)
				elog(ERROR, "failed to add item to index page in \"%s\"",
					 RelationGetRelationName(fc_index));

			memcpy(fc_ptr, fc_collector->tuples[fc_i], fc_tupsize);
			fc_ptr += fc_tupsize;

			fc_off++;
		}

		Assert((fc_ptr - fc_collectordata) <= fc_collector->sumsize);
		if (fc_needWal)
		{
			XLogRegisterBuffer(1, fc_buffer, REGBUF_STANDARD);
			XLogRegisterBufData(1, fc_collectordata, fc_collector->sumsize);
		}

		fc_metadata->tailFreeSize = PageGetExactFreeSpace(fc_page);

		MarkBufferDirty(fc_buffer);
	}

	/*
	 * 将 pd_lower 设置在元数据的末尾之后。 这是必不可少的，因为如果不这样做，如果 xlog.c 压缩页面，元数据将丢失。 （我们必须在这里做这件事，因为早于 v11 的 PG 版本没有正确设置元页面的 pd_lower，因此 pg_upgraded 索引可能包含错误的值。）
	 */
	((PageHeader) fc_metapage)->pd_lower =
		((char *) fc_metadata + sizeof(GinMetaPageData)) - (char *) fc_metapage;

	/*
	 * 写入元缓冲区，制作 xlog 条目
	 */
	MarkBufferDirty(fc_metabuffer);

	if (fc_needWal)
	{
		XLogRecPtr	fc_recptr;

		memcpy(&fc_data.metadata, fc_metadata, sizeof(GinMetaPageData));

		XLogRegisterBuffer(0, fc_metabuffer, REGBUF_WILL_INIT | REGBUF_STANDARD);
		XLogRegisterData((char *) &fc_data, sizeof(ginxlogUpdateMeta));

		fc_recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_UPDATE_META_PAGE);
		PageSetLSN(fc_metapage, fc_recptr);

		if (fc_buffer != InvalidBuffer)
		{
			PageSetLSN(fc_page, fc_recptr);
		}
	}

	if (fc_buffer != InvalidBuffer)
		UnlockReleaseBuffer(fc_buffer);

	/*
	 * 当待处理列表变得太长时强制清理。同时，
	 * ginInsertCleanup 可能会占用相当长的时间，因此我们
	 * 更倾向于在它可以在单个收集周期内完成所有工作时调用它。在
	 * 非清理模式下，不应该需要 maintenance_work_mem，因此在待处理列表
	 * 仍足够小以适合 gin_pending_list_limit 时触发它。
	 *
	 * ginInsertCleanup() 不应在我们的 CRIT_SECTION 内调用。
	 */
	fc_cleanupSize = GinGetPendingListCleanupSize(fc_index);
	if (fc_metadata->nPendingPages * GIN_PAGE_FREESIZE > fc_cleanupSize * 1024L)
		fc_needCleanup = true;

	UnlockReleaseBuffer(fc_metabuffer);

	END_CRIT_SECTION();

	/*
	 * 由于它可能与并发清理过程争用，我们不会强制清理
	 * 待处理列表。
	 */
	if (fc_needCleanup)
		ginInsertCleanup(fc_ginstate, false, true, false, NULL);
}

/*
 * 为单个可索引项创建临时索引元组（一个索引列
 * 对于由 ht_ctid 指定的堆元组），并将它们附加到
 * *collector 中的数组。它们将随后使用
 * ginHeapTupleFastInsert 写出。请注意，为了保证一致状态，
 * 针对给定堆元组的所有临时元组必须在一次调用中写入
 * ginHeapTupleFastInsert。
 */
void ginHeapTupleFastCollect(GinState *fc_ginstate,
						GinTupleCollector *fc_collector,
						OffsetNumber fc_attnum, Datum fc_value, bool fc_isNull,
						ItemPointer fc_ht_ctid)
{
	Datum	   *fc_entries;
	GinNullCategory *fc_categories;
	int32		fc_i,
				fc_nentries;

	/*
	 * 提取需要插入到索引中的键值
	 */
	fc_entries = ginExtractEntries(fc_ginstate, fc_attnum, fc_value, fc_isNull,
								&fc_nentries, &fc_categories);

	/*
	 * 在分配计算中防止整数溢出
	 */
	if (fc_nentries < 0 ||
		fc_collector->ntuples + fc_nentries > MaxAllocSize / sizeof(IndexTuple))
		elog(ERROR, "too many entries for GIN index");

	/*
	 * 为存储收集的元组分配/重新分配内存
	 */
	if (fc_collector->tuples == NULL)
	{
		/*
		 * 确定在元组数组中最初要分配的元素数量。
		 * 将其设置为2的幂，以避免在调整大小时
		 * 浪费内存（因为 palloc 喜欢2的幂）。
		 */
		fc_collector->lentuples = pg_nextpower2_32(Max(16, fc_nentries));
		fc_collector->tuples = (IndexTuple *) palloc(sizeof(IndexTuple) * fc_collector->lentuples);
	}
	else if (fc_collector->lentuples < fc_collector->ntuples + fc_nentries)
	{
		/*
		 * 将 lentuples 向前推进到下一个合适的2的幂。这不会
		 * 造成溢出，尽管我们可能会达到一个超过
		 * MaxAllocSize/sizeof(IndexTuple) 的值，从而导致 repalloc 出现错误。
		 */
		fc_collector->lentuples = pg_nextpower2_32(fc_collector->ntuples + fc_nentries);
		fc_collector->tuples = (IndexTuple *) repalloc(fc_collector->tuples,
													sizeof(IndexTuple) * fc_collector->lentuples);
	}

	/*
	 * 为每个键值构建一个索引元组，并添加到数组中。在待处理
	 * 元组中，我们只需将堆 TID 插入 t_tid。
	 */
	for (fc_i = 0; fc_i < fc_nentries; fc_i++)
	{
		IndexTuple	fc_itup;

		fc_itup = GinFormTuple(fc_ginstate, fc_attnum, fc_entries[fc_i], fc_categories[fc_i],
							NULL, 0, 0, true);
		fc_itup->t_tid = *fc_ht_ctid;
		fc_collector->tuples[fc_collector->ntuples++] = fc_itup;
		fc_collector->sumsize += IndexTupleSize(fc_itup);
	}
}

/*
 * 删除待处理列表页面，直到（不包括）newHead 页面。
 * 如果 newHead == InvalidBlockNumber，则函数删除整个列表。
 *
 * metapage 在此函数中被锁定并独占锁定。
 */
static void fc_shiftList(Relation fc_index, Buffer fc_metabuffer, BlockNumber fc_newHead,
		  bool fc_fill_fsm, IndexBulkDeleteResult *fc_stats)
{
	Page		fc_metapage;
	GinMetaPageData *fc_metadata;
	BlockNumber fc_blknoToDelete;

	fc_metapage = BufferGetPage(fc_metabuffer);
	fc_metadata = GinPageGetMeta(fc_metapage);
	fc_blknoToDelete = fc_metadata->head;

	do
	{
		Page		fc_page;
		int			fc_i;
		int64		fc_nDeletedHeapTuples = 0;
		ginxlogDeleteListPages fc_data;
		Buffer		fc_buffers[GIN_NDELETE_AT_ONCE];
		BlockNumber fc_freespace[GIN_NDELETE_AT_ONCE];

		fc_data.ndeleted = 0;
		while (fc_data.ndeleted < GIN_NDELETE_AT_ONCE && fc_blknoToDelete != fc_newHead)
		{
			fc_freespace[fc_data.ndeleted] = fc_blknoToDelete;
			fc_buffers[fc_data.ndeleted] = ReadBuffer(fc_index, fc_blknoToDelete);
			LockBuffer(fc_buffers[fc_data.ndeleted], GIN_EXCLUSIVE);
			fc_page = BufferGetPage(fc_buffers[fc_data.ndeleted]);

			fc_data.ndeleted++;

			Assert(!GinPageIsDeleted(fc_page));

			fc_nDeletedHeapTuples += GinPageGetOpaque(fc_page)->maxoff;
			fc_blknoToDelete = GinPageGetOpaque(fc_page)->rightlink;
		}

		if (fc_stats)
			fc_stats->pages_deleted += fc_data.ndeleted;

		/*
		 * 此操作会接触到异常数量的页面，因此
		 * 在进入关键区之前准备 XLogInsert 机制。
		 */
		if (RelationNeedsWAL(fc_index))
			XLogEnsureRecordSpace(fc_data.ndeleted, 0);

		START_CRIT_SECTION();

		fc_metadata->head = fc_blknoToDelete;

		Assert(fc_metadata->nPendingPages >= fc_data.ndeleted);
		fc_metadata->nPendingPages -= fc_data.ndeleted;
		Assert(fc_metadata->nPendingHeapTuples >= fc_nDeletedHeapTuples);
		fc_metadata->nPendingHeapTuples -= fc_nDeletedHeapTuples;

		if (fc_blknoToDelete == InvalidBlockNumber)
		{
			fc_metadata->tail = InvalidBlockNumber;
			fc_metadata->tailFreeSize = 0;
			fc_metadata->nPendingPages = 0;
			fc_metadata->nPendingHeapTuples = 0;
		}

		/*
		 * 将 pd_lower 设置在元数据末尾之后。这是必需的，
		 * 因为如果不这样做，元数据将在 xlog.c
		 * 压缩页面时丢失。（我们必须在这里这样做，因为在 v11
		 * 之前的 PG 版本没有正确设置元页面的 pd_lower，因此一个
		 * pg_upgraded 索引可能包含错误的值。）
		 */
		((PageHeader) fc_metapage)->pd_lower =
			((char *) fc_metadata + sizeof(GinMetaPageData)) - (char *) fc_metapage;

		MarkBufferDirty(fc_metabuffer);

		for (fc_i = 0; fc_i < fc_data.ndeleted; fc_i++)
		{
			fc_page = BufferGetPage(fc_buffers[fc_i]);
			GinPageGetOpaque(fc_page)->flags = GIN_DELETED;
			MarkBufferDirty(fc_buffers[fc_i]);
		}

		if (RelationNeedsWAL(fc_index))
		{
			XLogRecPtr	fc_recptr;

			XLogBeginInsert();
			XLogRegisterBuffer(0, fc_metabuffer,
							   REGBUF_WILL_INIT | REGBUF_STANDARD);
			for (fc_i = 0; fc_i < fc_data.ndeleted; fc_i++)
				XLogRegisterBuffer(fc_i + 1, fc_buffers[fc_i], REGBUF_WILL_INIT);

			memcpy(&fc_data.metadata, fc_metadata, sizeof(GinMetaPageData));

			XLogRegisterData((char *) &fc_data,
							 sizeof(ginxlogDeleteListPages));

			fc_recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_DELETE_LISTPAGE);
			PageSetLSN(fc_metapage, fc_recptr);

			for (fc_i = 0; fc_i < fc_data.ndeleted; fc_i++)
			{
				fc_page = BufferGetPage(fc_buffers[fc_i]);
				PageSetLSN(fc_page, fc_recptr);
			}
		}

		for (fc_i = 0; fc_i < fc_data.ndeleted; fc_i++)
			UnlockReleaseBuffer(fc_buffers[fc_i]);

		END_CRIT_SECTION();

		for (fc_i = 0; fc_fill_fsm && fc_i < fc_data.ndeleted; fc_i++)
			RecordFreeIndexPage(fc_index, fc_freespace[fc_i]);

	} while (fc_blknoToDelete != fc_newHead);
}

/* 初始化空的 KeyArray */
static void fc_initKeyArray(KeyArray *fc_keys, int32 fc_maxvalues)
{
	fc_keys->keys = (Datum *) palloc(sizeof(Datum) * fc_maxvalues);
	fc_keys->categories = (GinNullCategory *)
		palloc(sizeof(GinNullCategory) * fc_maxvalues);
	fc_keys->nvalues = 0;
	fc_keys->maxvalues = fc_maxvalues;
}

/* 将数据添加到 KeyArray，如有必要则调整大小 */
static void fc_addDatum(KeyArray *fc_keys, Datum fc_datum, GinNullCategory fc_category)
{
	if (fc_keys->nvalues >= fc_keys->maxvalues)
	{
		fc_keys->maxvalues *= 2;
		fc_keys->keys = (Datum *)
			repalloc(fc_keys->keys, sizeof(Datum) * fc_keys->maxvalues);
		fc_keys->categories = (GinNullCategory *)
			repalloc(fc_keys->categories, sizeof(GinNullCategory) * fc_keys->maxvalues);
	}

	fc_keys->keys[fc_keys->nvalues] = fc_datum;
	fc_keys->categories[fc_keys->nvalues] = fc_category;
	fc_keys->nvalues++;
}

/*
 * 从待处理列表页面收集数据，以便插入到
 * 主要索引中。
 *
 * 遍历页面上所有 >= startoff 的元组并收集值到 accum
 *
 * 请注意，ka 只是工作空间——它不会在
 * 调用之间携带任何状态。
 */
static void fc_processPendingPage(BuildAccumulator *fc_accum, KeyArray *fc_ka,
				   Page fc_page, OffsetNumber fc_startoff)
{
	ItemPointerData fc_heapptr;
	OffsetNumber fc_i,
				fc_maxoff;
	OffsetNumber fc_attrnum;

	/* 重置 *ka 为空 */
	fc_ka->nvalues = 0;

	fc_maxoff = PageGetMaxOffsetNumber(fc_page);
	Assert(fc_maxoff >= FirstOffsetNumber);
	ItemPointerSetInvalid(&fc_heapptr);
	fc_attrnum = 0;

	for (fc_i = fc_startoff; fc_i <= fc_maxoff; fc_i = OffsetNumberNext(fc_i))
	{
		IndexTuple	fc_itup = (IndexTuple) PageGetItem(fc_page, PageGetItemId(fc_page, fc_i));
		OffsetNumber fc_curattnum;
		Datum		fc_curkey;
		GinNullCategory fc_curcategory;

		/* 检查堆 TID 或 attnum 是否变化 */
		fc_curattnum = gintuple_get_attrnum(fc_accum->ginstate, fc_itup);

		if (!ItemPointerIsValid(&fc_heapptr))
		{
			fc_heapptr = fc_itup->t_tid;
			fc_attrnum = fc_curattnum;
		}
		else if (!(ItemPointerEquals(&fc_heapptr, &fc_itup->t_tid) &&
				   fc_curattnum == fc_attrnum))
		{
			/*
			 * ginInsertBAEntries 每次调用可以插入多个数据，但仅
			 * 针对一个堆元组和一列。因此在边界调用它，
			 * 并重置 ka。
			 */
			ginInsertBAEntries(fc_accum, &fc_heapptr, fc_attrnum,
							   fc_ka->keys, fc_ka->categories, fc_ka->nvalues);
			fc_ka->nvalues = 0;
			fc_heapptr = fc_itup->t_tid;
			fc_attrnum = fc_curattnum;
		}

		/* 将键添加到 KeyArray */
		fc_curkey = gintuple_get_key(fc_accum->ginstate, fc_itup, &fc_curcategory);
		fc_addDatum(fc_ka, fc_curkey, fc_curcategory);
	}

	/* 转储所有剩余键 */
	ginInsertBAEntries(fc_accum, &fc_heapptr, fc_attrnum,
					   fc_ka->keys, fc_ka->categories, fc_ka->nvalues);
}

/*
 * 将元组从待处理页面移入常规 GIN 结构。
 *
 * 乍一看，它似乎完全不具备崩溃安全性。但如果在将条目发布到主索引后并在将它们从待处理列表中移除之前崩溃，那也是可以的，因为当我们稍后重新执行发布时，不会发生任何不好的事情。
 *
 * fill_fsm 表示 ginInsertCleanup 应该将已删除的页面添加到 FSM，否则调用者负责将已删除的页面放入 FSM。
 *
 * 如果 stats 不是 null，我们会将已删除的待处理页面计入统计。
 */
void ginInsertCleanup(GinState *fc_ginstate, bool fc_full_clean,
				 bool fc_fill_fsm, bool fc_forceCleanup,
				 IndexBulkDeleteResult *fc_stats)
{
	Relation	fc_index = fc_ginstate->index;
	Buffer		fc_metabuffer,
				fc_buffer;
	Page		fc_metapage,
				fc_page;
	GinMetaPageData *fc_metadata;
	MemoryContext fc_opCtx,
				fc_oldCtx;
	BuildAccumulator fc_accum;
	KeyArray	fc_datums;
	BlockNumber fc_blkno,
				fc_blknoFinish;
	bool		fc_cleanupFinish = false;
	bool		fc_fsm_vac = false;
	Size		fc_workMemory;

	/*
	 * 我们希望防止并发清理过程。为此，我们将使用 LockPage() 调用以独占模式锁定元页面。没有其他人会使用该锁定来处理元页面，因此我们保留了将条目并发插入到待处理列表中的可能性。
	 */

	if (fc_forceCleanup)
	{
		/*
		 * 我们是从 [自动] 真空/分析或 gin_clean_pending_list() 被调用的，想要等待并发清理完成。
		 */
		LockPage(fc_index, GIN_METAPAGE_BLKNO, ExclusiveLock);
		fc_workMemory =
			(IsAutoVacuumWorkerProcess() && autovacuum_work_mem != -1) ?
			autovacuum_work_mem : maintenance_work_mem;
	}
	else
	{
		/*
		 * 我们是从常规插入被调用的，如果我们看到并发清理就直接退出，希望并发进程会清理待处理列表。
		 */
		if (!ConditionalLockPage(fc_index, GIN_METAPAGE_BLKNO, ExclusiveLock))
			return;
		fc_workMemory = work_mem;
	}

	fc_metabuffer = ReadBuffer(fc_index, GIN_METAPAGE_BLKNO);
	LockBuffer(fc_metabuffer, GIN_SHARE);
	fc_metapage = BufferGetPage(fc_metabuffer);
	fc_metadata = GinPageGetMeta(fc_metapage);

	if (fc_metadata->head == InvalidBlockNumber)
	{
		/* 无需执行任何操作 */
		UnlockReleaseBuffer(fc_metabuffer);
		UnlockPage(fc_index, GIN_METAPAGE_BLKNO, ExclusiveLock);
		return;
	}

	/*
	 * 记住一个尾页面，以防其他后端以更快的速度添加新元组，导致无限清理。
	 */
	fc_blknoFinish = fc_metadata->tail;

	/*
	 * 读取并锁定待处理列表的头部
	 */
	fc_blkno = fc_metadata->head;
	fc_buffer = ReadBuffer(fc_index, fc_blkno);
	LockBuffer(fc_buffer, GIN_SHARE);
	fc_page = BufferGetPage(fc_buffer);

	LockBuffer(fc_metabuffer, GIN_UNLOCK);

	/*
	 * 初始化。所有临时空间将位于 opCtx 中
	 */
	fc_opCtx = AllocSetContextCreate(CurrentMemoryContext,
								  "GIN insert cleanup temporary context",
								  ALLOCSET_DEFAULT_SIZES);

	fc_oldCtx = MemoryContextSwitchTo(fc_opCtx);

	fc_initKeyArray(&fc_datums, 128);
	ginInitBA(&fc_accum);
	fc_accum.ginstate = fc_ginstate;

	/*
	 * 在此循环的顶部，我们锁定当前待处理列表的页面。但是，在退出循环之前，我们将释放锁。请注意，我们也锁定了元页面，但不是锁定状态。
	 */
	for (;;)
	{
		Assert(!GinPageIsDeleted(fc_page));

		/*
		 * 我们是否正在遍历页面，而我们记得在开始清理时它是一个尾页面？但是如果调用者要求我们清理整个待处理列表，就忽略旧尾部，我们将一直处理直到列表变为空。
		 */
		if (fc_blkno == fc_blknoFinish && fc_full_clean == false)
			fc_cleanupFinish = true;

		/*
		 * 将页面的数据读入累加器
		 */
		fc_processPendingPage(&fc_accum, &fc_datums, fc_page, FirstOffsetNumber);

		vacuum_delay_point();

		/*
		 * 是时候将内存刷新到磁盘了吗？如果我们在待处理列表的末尾，或者如果我们有一整行且内存快满了，就刷新。
		 */
		if (GinPageGetOpaque(fc_page)->rightlink == InvalidBlockNumber ||
			(GinPageHasFullRow(fc_page) &&
			 (fc_accum.allocatedMemory >= fc_workMemory * 1024L)))
		{
			ItemPointerData *fc_list;
			uint32		fc_nlist;
			Datum		fc_key;
			GinNullCategory fc_category;
			OffsetNumber fc_maxoff,
						fc_attnum;

			/*
			 * 解锁当前页面以提高性能。页面的变化将在内存刷新完成后通过比较 maxoff 来检查。
			 */
			fc_maxoff = PageGetMaxOffsetNumber(fc_page);
			LockBuffer(fc_buffer, GIN_UNLOCK);

			/*
			 * 将收集到的数据移入常规结构可能需要相当长的时间 - 因此，在不锁定待处理列表的情况下运行它。
			 */
			ginBeginBAScan(&fc_accum);
			while ((fc_list = ginGetBAEntry(&fc_accum,
										 &fc_attnum, &fc_key, &fc_category, &fc_nlist)) != NULL)
			{
				ginEntryInsert(fc_ginstate, fc_attnum, fc_key, fc_category,
							   fc_list, fc_nlist, NULL);
				vacuum_delay_point();
			}

			/*
			 * 锁定整个列表以移除页面
			 */
			LockBuffer(fc_metabuffer, GIN_EXCLUSIVE);
			LockBuffer(fc_buffer, GIN_SHARE);

			Assert(!GinPageIsDeleted(fc_page));

			/*
			 * 当我们将页面解锁时，可能有更多内容被添加到其中。如果是这样，则立即处理这些条目。数量不应该很多，因此我们不担心使用独占锁的事实。插入算法保证插入的行不会继续在下一页。
			 * 注意：在这个循环中故意没有 vacuum_delay_point。
			 */
			if (PageGetMaxOffsetNumber(fc_page) != fc_maxoff)
			{
				ginInitBA(&fc_accum);
				fc_processPendingPage(&fc_accum, &fc_datums, fc_page, fc_maxoff + 1);

				ginBeginBAScan(&fc_accum);
				while ((fc_list = ginGetBAEntry(&fc_accum,
											 &fc_attnum, &fc_key, &fc_category, &fc_nlist)) != NULL)
					ginEntryInsert(fc_ginstate, fc_attnum, fc_key, fc_category,
								   fc_list, fc_nlist, NULL);
			}

			/*
			 * 记住下一页面 - 它将成为新的列表头
			 */
			fc_blkno = GinPageGetOpaque(fc_page)->rightlink;
			UnlockReleaseBuffer(fc_buffer);	/* shiftList 将执行独占
											 * 锁定 */

			/*
			 * 从待处理列表中移除已读取的页面，此时已读取页面的所有内容
			 * 在常规结构中
			 */
			fc_shiftList(fc_index, fc_metabuffer, fc_blkno, fc_fill_fsm, fc_stats);

			/* 此时，一些待处理页面已经被释放 */
			fc_fsm_vac = true;

			Assert(fc_blkno == fc_metadata->head);
			LockBuffer(fc_metabuffer, GIN_UNLOCK);

			/*
			 * 如果我们移除了整个待处理列表，或者我们清理了尾部（我们
			 * 在开始清理过程时记住的）那么就直接退出
			 */
			if (fc_blkno == InvalidBlockNumber || fc_cleanupFinish)
				break;

			/*
			 * 释放迄今为止使用的内存并重新初始化状态
			 */
			MemoryContextReset(fc_opCtx);
			fc_initKeyArray(&fc_datums, fc_datums.maxvalues);
			ginInitBA(&fc_accum);
		}
		else
		{
			fc_blkno = GinPageGetOpaque(fc_page)->rightlink;
			UnlockReleaseBuffer(fc_buffer);
		}

		/*
		 * 在待处理列表中读取下一页
		 */
		vacuum_delay_point();
		fc_buffer = ReadBuffer(fc_index, fc_blkno);
		LockBuffer(fc_buffer, GIN_SHARE);
		fc_page = BufferGetPage(fc_buffer);
	}

	UnlockPage(fc_index, GIN_METAPAGE_BLKNO, ExclusiveLock);
	ReleaseBuffer(fc_metabuffer);

	/*
	 * 由于待处理列表页面可能具有高的更换率，因此希望在普通后端
	 * 清理列表时立即将其回收到FreeSpaceMap中。
	 */
	if (fc_fsm_vac && fc_fill_fsm)
		IndexFreeSpaceMapVacuum(fc_index);

	/* 清理临时空间 */
	MemoryContextSwitchTo(fc_oldCtx);
	MemoryContextDelete(fc_opCtx);
}

/*
 * 可通过SQL调用的函数，以清理插入待处理列表
 */
Datum gin_clean_pending_list(PG_FUNCTION_ARGS)
{
	Oid			fc_indexoid = PG_GETARG_OID(0);
	Relation	fc_indexRel = index_open(fc_indexoid, RowExclusiveLock);
	IndexBulkDeleteResult fc_stats;

	if (RecoveryInProgress())
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("recovery is in progress"),
				 errhint("GIN pending list cannot be cleaned up during recovery.")));

	/* 必须是GIN索引 */
	if (fc_indexRel->rd_rel->relkind != RELKIND_INDEX ||
		fc_indexRel->rd_rel->relam != GIN_AM_OID)
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" is not a GIN index",
						RelationGetRelationName(fc_indexRel))));

	/*
	 * 拒绝尝试读取非本地临时关系；我们可能会获取错误的数据，因为
	 * 我们无法查看拥有会话的本地缓冲区。
	 */
	if (RELATION_IS_OTHER_TEMP(fc_indexRel))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot access temporary indexes of other sessions")));

	/* 用户必须拥有索引（与 VACUUM 所需的权限相当） */
	if (!pg_class_ownercheck(fc_indexoid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_INDEX,
					   RelationGetRelationName(fc_indexRel));

	memset(&fc_stats, 0, sizeof(fc_stats));

	/*
	 * 无法对!indisready索引的内容做出任何假设。将其视为无操作，
	 * 而不是错误，这样用户可以在所有访问方法的索引上运行此函数。
	 * 由于一个indisready && !indisvalid索引仅在等待遗漏的aminsert
	 * 调用，因此我们能够处理它。由于过度谨慎，拒绝这样做。
	 */
	if (fc_indexRel->rd_index->indisvalid)
	{
		GinState	fc_ginstate;

		initGinState(&fc_ginstate, fc_indexRel);
		ginInsertCleanup(&fc_ginstate, true, true, true, &fc_stats);
	}
	else
		ereport(DEBUG1,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("index \"%s\" is not valid",
						RelationGetRelationName(fc_indexRel))));

	index_close(fc_indexRel, RowExclusiveLock);

	PG_RETURN_INT64((int64) fc_stats.pages_deleted);
}
